#include <memory>
#include <iostream>
#include <brpc/server.h>
#include <butil/logging.h>
#include<unordered_set>

// common文件夹下
#include "data_es.hpp"
#include "mysql_message.hpp"
#include "etcder.hpp"
#include "uuid.hpp"
#include "channel.hpp"
#include "logger.hpp"
#include "rabbitmq.hpp"

#include "message.pb.h"
#include "base.pb.h"
#include "user.pb.h"
#include "file.pb.h"
#include "message.hxx"
namespace bite_im
{
    // 实现用户模块功能

    // 用户模块服务类
    class MessageServiceImpl : public bite_im::MsgStorageService
    {
    public:
        using ptr = std::shared_ptr<MessageServiceImpl>;
        MessageServiceImpl(const ElasticPtr &es_client,
                           const DBPtr &mysql_client,
                           const ServiceManager::ptr &service_manager,
                           const std::string &file_service_name,
                           const std::string &user_service_name)
            : _es_message(std::make_shared<ESMessage>(es_client)),
              _mysql_message(std::make_shared<MessageTable>(mysql_client)),
              _service_manager(service_manager),
              _file_service_name(file_service_name),
              _user_service_name(user_service_name)
        {
        }

        ~MessageServiceImpl() {}

        /* --------------------------------------------- 业务函数----------------------------------------------------*/

        // 获取历史消息
        void GetHistoryMsg(google::protobuf::RpcController *controller,
                           const ::bite_im::GetHistoryMsgReq *request,
                           ::bite_im::GetHistoryMsgRsp *response,
                           ::google::protobuf::Closure *done)
        {
            // 错误函数---lambda
            auto err_response_func = [this, response](const std::string &request_id,
                                                      const std::string &errmsg) -> void
            {
                response->set_request_id(request_id);
                response->set_success(false);
                response->set_errmsg(errmsg);
                return;
            };
            // 管理rpc对象
            brpc::ClosureGuard rpc_guard(done);

            // 1.获取会话id，起始时间，结束时间
            std::string chat_session_id = request->chat_session_id();
            boost::posix_time::ptime start_time(boost::posix_time::from_time_t(request->start_time()));
            boost::posix_time::ptime end_time(boost::posix_time::from_time_t(request->over_time()));

            // 2.从数据库中进行消息查询，并获取消息
            std::vector<Message> message_list = _mysql_message->range(chat_session_id, start_time, end_time);

            // 3.统计消息所以文件类型消息的文件id，调用文件子服务进行下载

            std::unordered_set<std::string> file_id_list;   // 不能够用数组进行存储，一个人可以发多条信息，如果重复了最后sql查询是会多查询

            for (const auto &message : message_list)
            {
                if (message.file_id().empty() == false)
                {
                    file_id_list.insert(message.file_id());
                }
            }
            // 文件id-文件数据映射哈希表
            std::unordered_map<std::string, std::string> file_data_map;
            // 进行文件子服务下载批量文件，并通过file_data_map读取到文件数据
            bool ret = ReadFile(file_id_list, file_data_map);
            if (ret == false)
            {
                LOG_ERROR("进行批量化下载历史文件失败，请求id[]", request->request_id());
                return err_response_func(request->request_id(), "进行批量化下载历史文件失败");
            }

            // 4.统计所以消息发送者的用户id，从用户子服务进行批量用户信息获取
            std::unordered_set<std::string> user_id_list;
            for (const auto &message : message_list)
            {
                user_id_list.insert(message.user_id());
            }
            std::unordered_map<std::string, UserInfo> user_data_map;
            ret = GetUser(user_id_list, user_data_map);
            if (ret == false)
            {
                LOG_ERROR("进行批量化获取用户失败，请求id[]", request->request_id());
                return err_response_func(request->request_id(), "进行批量化获取用户失败");
            }

            // 此时我们就有了当前聊天中 所有文件数据file_data_map 和 用户信息数据user_data_map

            // 5.组织响应
            response->set_request_id(request->request_id());
            response->set_success(true);
            // 组织repeated MessageInfo msg_list
            for (const auto &message : message_list)
            {
                // 这个接口特殊处理了，返回指针，我们对指针进行操作，就相当于将message_info插入进
                auto message_info = response->add_msg_list();
                message_info->set_message_id(message.message_id());
                message_info->set_chat_session_id(message.session_id());
                message_info->set_timestamp(boost::posix_time::to_time_t(message.create_time()));
                message_info->mutable_sender()->CopyFrom(user_data_map[message.user_id()]); // 通过用户名在user_data_map获取用户信息数据

                // 针对文件类型进行消息内容接口的填充
                switch (message.message_type())
                {
                case MessageType::STRING:
                {
                    message_info->mutable_message()->set_message_type(MessageType::STRING);
                    message_info->mutable_message()->mutable_string_message()->set_content(message.content());
                }
                break;
                case MessageType::IMAGE:
                {
                    message_info->mutable_message()->set_message_type(MessageType::IMAGE);
                    message_info->mutable_message()->mutable_image_message()->set_file_id(message.file_id());
                    message_info->mutable_message()->mutable_image_message()->set_image_content(file_data_map[message.file_id()]);
                }
                break;
                case MessageType::FILE:
                {
                    message_info->mutable_message()->set_message_type(MessageType::FILE);
                    message_info->mutable_message()->mutable_file_message()->set_file_id(message.file_id());
                    message_info->mutable_message()->mutable_file_message()->set_file_size(message.file_size());
                    message_info->mutable_message()->mutable_file_message()->set_file_name(message.file_name());
                    message_info->mutable_message()->mutable_file_message()->set_file_contents(file_data_map[message.file_id()]);
                }
                break;
                case MessageType::SPEECH:
                {
                    message_info->mutable_message()->set_message_type(MessageType::SPEECH);
                    message_info->mutable_message()->mutable_speech_message()->set_file_id(message.file_id());
                    message_info->mutable_message()->mutable_speech_message()->set_file_contents(file_data_map[message.file_id()]);
                }
                break;
                default:
                    LOG_ERROR("文件类型[{}]错误", message.message_type());
                    return;
                }
            }
        }

        void GetRecentMsg(google::protobuf::RpcController *controller,
                          const ::bite_im::GetRecentMsgReq *request,
                          ::bite_im::GetRecentMsgRsp *response,
                          ::google::protobuf::Closure *done)
        {
            // 错误函数---lambda
            auto err_response_func = [this, response](const std::string &request_id,
                                                      const std::string &errmsg) -> void
            {
                response->set_request_id(request_id);
                response->set_success(false);
                response->set_errmsg(errmsg);
                return;
            };
            // 管理rpc对象
            brpc::ClosureGuard rpc_guard(done);
            // 1.获取会话id，消息数目
            std::string chat_session_id = request->chat_session_id();
            int count = request->msg_count();

            // 2.从数据库中进行消息查询，并获取消息
            std::vector<Message> message_list = _mysql_message->recent(chat_session_id, count);

            // 3.统计消息所以文件类型消息的文件id，调用文件子服务进行下载
            std::unordered_set<std::string> file_id_list;
            for (const auto &message : message_list)
            {
                if (message.file_id().empty() == false)
                {
                    file_id_list.insert(message.file_id());
                }
            }
            // 文件id-文件数据映射哈希表
            std::unordered_map<std::string, std::string> file_data_map;
            // 进行文件子服务下载批量文件，并通过file_data_map读取到文件数据
            bool ret = ReadFile(file_id_list, file_data_map);
            if (ret == false)
            {
                LOG_ERROR("进行批量化下载历史文件失败，请求id[]", request->request_id());
                return err_response_func(request->request_id(), "进行批量化下载历史文件失败");
            }

            // 4.统计所以消息发送者的用户id，从用户子服务进行批量用户信息获取
            std::unordered_set<std::string> user_id_list;
            for (const auto &message : message_list)
            {
                user_id_list.insert(message.user_id());
            }
            std::unordered_map<std::string, UserInfo> user_data_map;
            ret = GetUser(user_id_list, user_data_map);
            if (ret == false)
            {
                LOG_ERROR("进行批量化获取用户失败，请求id[]", request->request_id());
                return err_response_func(request->request_id(), "进行批量化获取用户失败");
            }

            // 此时我们就有了当前聊天中 所有文件数据file_data_map 和 用户信息数据user_data_map

            // 5.组织响应
            response->set_request_id(request->request_id());
            response->set_success(true);
                       // 组织repeated MessageInfo msg_list
            for (const auto &message : message_list)
            {
                // 这个接口特殊处理了，返回指针，我们对指针进行操作，就相当于将message_info插入进
                auto message_info = response->add_msg_list();
                message_info->set_message_id(message.message_id());
                message_info->set_chat_session_id(message.session_id());
                message_info->set_timestamp(boost::posix_time::to_time_t(message.create_time()));
                message_info->mutable_sender()->CopyFrom(user_data_map[message.user_id()]); // 通过用户名在user_data_map获取用户信息数据

                // 针对文件类型进行消息内容接口的填充
                switch (message.message_type())
                {
                case MessageType::STRING:
                {
                    message_info->mutable_message()->set_message_type(MessageType::STRING);
                    message_info->mutable_message()->mutable_string_message()->set_content(message.content());
                }
                break;
                case MessageType::IMAGE:
                {
                    message_info->mutable_message()->set_message_type(MessageType::IMAGE);
                    message_info->mutable_message()->mutable_image_message()->set_file_id(message.file_id());
                    message_info->mutable_message()->mutable_image_message()->set_image_content(file_data_map[message.file_id()]);
                }
                break;
                case MessageType::FILE:
                {
                    message_info->mutable_message()->set_message_type(MessageType::FILE);
                    message_info->mutable_message()->mutable_file_message()->set_file_id(message.file_id());
                    message_info->mutable_message()->mutable_file_message()->set_file_size(message.file_size());
                    message_info->mutable_message()->mutable_file_message()->set_file_name(message.file_name());
                    message_info->mutable_message()->mutable_file_message()->set_file_contents(file_data_map[message.file_id()]);
                }
                break;
                case MessageType::SPEECH:
                {
                    message_info->mutable_message()->set_message_type(MessageType::SPEECH);
                    message_info->mutable_message()->mutable_speech_message()->set_file_id(message.file_id());
                    message_info->mutable_message()->mutable_speech_message()->set_file_contents(file_data_map[message.file_id()]);
                }
                break;
                default:
                    LOG_ERROR("文件类型[{}]错误", message.message_type());
                    return;
                }
            }
        }

        void MsgSearch(google::protobuf::RpcController *controller,
                       const ::bite_im::MsgSearchReq *request,
                       ::bite_im::MsgSearchRsp *response,
                       ::google::protobuf::Closure *done)
        {
            // 错误函数---lambda
            auto err_response_func = [this, response](const std::string &request_id,
                                                      const std::string &errmsg) -> void
            {
                response->set_request_id(request_id);
                response->set_success(false);
                response->set_errmsg(errmsg);
                return;
            };
            // 管理rpc对象
            brpc::ClosureGuard rpc_guard(done);

            // 1.从es模块获取查询的文本消息

            // 消息的查询是只查询文本消息，因为我们在es中可以实现关键字查询
            std::string chat_session_id = request->chat_session_id();
            std::string search_key = request->search_key();
            auto message_list = _es_message->search(search_key, chat_session_id);
            LOG_INFO("消息数目: {}", message_list.size());
            // 2.统计所有消息发送者的用户id，从用户子服务进行批量用户信息获取
            std::unordered_set<std::string> user_id_list;
            for (const auto &message : message_list)
            {
                user_id_list.insert(message.user_id());
            }
            std::unordered_map<std::string, UserInfo> user_data_map;
            bool ret = GetUser(user_id_list, user_data_map);
            if (ret == false)
            {
                LOG_ERROR("进行批量化获取用户失败，请求id[]", request->request_id());
                return err_response_func(request->request_id(), "进行批量化获取用户失败");
            }

            // 此时我们就有了当前聊天中用户信息数据user_data_map

            // 5.组织响应
            response->set_request_id(request->request_id());
            response->set_success(true);
            // 组织repeated MessageInfo msg_list
            for (const auto &message : message_list)
            {
                // 这个接口特殊处理了，返回指针，我们对指针进行操作，就相当于将message_info插入进msg_list
                auto message_info = response->add_msg_list();
                message_info->set_message_id(message.message_id());
                message_info->set_chat_session_id(message.session_id());
                message_info->set_timestamp(boost::posix_time::to_time_t(message.create_time()));
                message_info->mutable_sender()->CopyFrom(user_data_map[message.user_id()]); // 通过用户名在user_data_map获取用户信息数据
                message_info->mutable_message()->set_message_type(MessageType::STRING);
                message_info->mutable_message()->mutable_string_message()->set_content(message.content());
            }
        }

        // 当我们在消息转发子服务中发布一条消息给消息队列，接着就是消息存储子服务提供消息队列的消费函数，后续就是消息子服务进行消息存储
        // 消息队列消费回调处理函数---进入时已经获取到了文件信息
        // 当mq_client获得到一条新消息进行消费，会调用回调函数
        void onMessage(const char *body, size_t size)
        {
            // 1.取出网络中序列化的消息内容，进行反序列化获取消息结构体

            bite_im::MessageInfo message_info; // 消息结构体字段 ，如果冲突就换一个变量名message_Info
            bool ret = message_info.ParseFromArray(body, size);
            if (ret == false)
            {
                LOG_ERROR("对消息队列的消息进行反序列化失败");
                return;
            }

            // 2.根据不同的消息类型进行不同的操作

            // 消息内容字段
            bite_im::MessageContent message_content = message_info.message();
            std::string text_content; // 只有文本消息才会进行存储，不然为空
            std::string file_id;
            std::string file_name;
            int64_t file_size;
            switch (message_content.message_type())
            {
            // 如果为文本消息，那么需要存入es中
            case MessageType::STRING:
            {
                // 获取文本消息并存入到es中
                auto message = message_content.string_message();
                // 存储一份文本消息，来添加到mysql中
                text_content = message.content();
                ret = _es_message->appendData(message_info.sender().user_id(),
                                              message_info.message_id(),
                                              message_info.timestamp(),
                                              message_info.chat_session_id(),
                                              message.content());
                if (ret == false)
                {
                    LOG_ERROR("添加文本消息到ES搜索引擎新失败");
                    return;
                }
            }
            break;
            case MessageType::IMAGE:
            {
                auto message = message_content.image_message();
                ret = PutFile("", message.image_content(), message.image_content().size(), file_id);
                if (ret == false)
                {
                    LOG_ERROR("上传图像到文件子服务失败");
                    return;
                }
            }
            break;
            case MessageType::FILE:
            {
                auto message = message_content.file_message();
                file_name = message.file_name();
                file_size = message.file_size();
                ret = PutFile(message.file_name(), message.file_contents(), message.file_size(), file_id);
                if (ret == false)
                {
                    LOG_ERROR("上传文件到文件子服务失败");
                    return;
                }
            }
            break;
            case MessageType::SPEECH:
            {
                auto message = message_content.speech_message();
                ret = PutFile("", message.file_contents(), message.file_contents().size(), file_id);
                if (ret == false)
                {
                    LOG_ERROR("上传语音到文件子服务失败");
                    return;
                }
            }
            break;
            default:
                LOG_ERROR("文件类型[{}]错误", message_content.message_type());
                return;
            }

            // 3.提取消息元消息，存储到mysql数据库中

            boost::posix_time::ptime ctime(boost::posix_time::from_time_t(message_info.timestamp()));
            // 组织Message存储结构
            bite_im::Message message(message_info.message_id(),
                                     message_info.chat_session_id(),
                                     message_info.sender().user_id(),
                                     message_info.message().message_type(),
                                     ctime);
            message.content(text_content); // 如果是文本消息才提供该字段
            message.file_id(file_id);
            message.file_name(file_name);
            message.file_size(file_size);
            ret = _mysql_message->insert(message);
            if (ret == false)
            {
                LOG_ERROR("向数据库存储消费的新消息失败");
                return;
            }
        }

    private:
        bool GetUser(const std::unordered_set<std::string> &user_id_list,
                     std::unordered_map<std::string, UserInfo> &user_data_map)
        {
            // 2.构建用户子服务
            auto channel = _service_manager->choose(_user_service_name);
            if (!channel) // 服务不存在
            {
                LOG_ERROR("用户子服务[{}]节点未找到", _user_service_name);
                return false;
            }
            bite_im::UserService_Stub stub(channel.get());
            // 3.获取消息发送的发送者用户信息
            bite_im::GetMultiUserInfoReq req;
            req.set_request_id(UUID::uuid());
            for (const auto &user_id : user_id_list)
            {
                req.add_users_id(user_id);
            }
            GetMultiUserInfoRsp resp;
            brpc::Controller cntl;
            stub.GetMultiUserInfo(&cntl, &req, &resp, nullptr);
            if (cntl.Failed() == true || resp.success() == false)
            {
                LOG_ERROR("用户子服务调用失败]，原因[{}]", cntl.ErrorText());
                return false;
            }
            const auto &info_map = resp.users_info();

            for (const auto &info : info_map)
            {
                user_data_map[info.first] = info.second;
            }
            return true;
        }

        bool ReadFile(const std::unordered_set<std::string> &file_id_list,
                      std::unordered_map<std::string, std::string> &file_data_map)
        {
            // 进行文件子服务下载批量文科
            auto channel = _service_manager->choose(_file_service_name);
            if (!channel) // 服务不存在
            {
                LOG_ERROR("文件子服务[{}]节点未找到", _file_service_name);
                return false;
            }
            // 获取到文件子服务操作对象
            bite_im::FileService_Stub stub(channel.get());
            bite_im::GetMultiFileReq req;
            // 设置文件子服务的请求
            req.set_request_id(UUID::uuid());
            for (const auto &file_id : file_id_list)
            {
                req.add_file_id_list(file_id);
            }
            bite_im::GetMultiFileRsp resp;
            // 获取一个rpc操作对象
            brpc::Controller cntl;
            // stub调用我们实现的服务类ipml的接口
            stub.GetMultiFile(&cntl, &req, &resp, nullptr);
            if (cntl.Failed() == true || resp.success() == false)
            {
                LOG_ERROR("文件子服务调用失败，原因[{}]", cntl.ErrorText());
                return false;
            }
            // data_map为map<string, FileDownloadData> file_data
            const auto &data_map = resp.file_data();
            for (auto data : data_map)
            {
                // first为文件id，second为FileDownloadData
                file_data_map[data.first] = data.second.file_content();
            }
            return true;
        }

        // 实现文件数据的上传
        bool PutFile(const std::string &file_name,
                     const std::string &body,
                     int64_t file_size,
                     std::string &file_id)
        {
            // 从服务信道管理模块获取文件存储子服务的service
            auto channel = _service_manager->choose(_file_service_name);
            if (!channel) // 服务不存在
            {
                LOG_ERROR("文件子服务[{}]节点未找到", _file_service_name);
                return false;
            }
            // 获取到文件子服务操作对象
            bite_im::FileService_Stub stub(channel.get());
            bite_im::PutSingleFileReq req;
            // 设置文件子服务的请求
            req.set_request_id(UUID::uuid());
            req.mutable_file_data()->set_file_name(file_name);
            req.mutable_file_data()->set_file_size(file_size);
            req.mutable_file_data()->set_file_content(body);

            bite_im::PutSingleFileRsp resp;
            // 获取一个rpc操作对象
            brpc::Controller cntl;
            // stub调用我们实现的服务类ipml的接口
            stub.PutSingleFile(&cntl, &req, &resp, nullptr);
            if (cntl.Failed() == true || resp.success() == false)
            {
                LOG_ERROR("文件子服务调用失败，原因[{}]", cntl.ErrorText());
                return false;
            }
            // 外部就能够得到文件id了
            file_id = resp.file_info().file_id();
            return true;
        }

    private:
        // es存储
        ESMessage::ptr _es_message;
        // mysql存储
        MessageTable::ptr _mysql_message;

        // rpc获取关心服务的存储服务信道
        std::string _file_service_name;
        std::string _user_service_name;
        ServiceManager::ptr _service_manager;
    };

    class MessageServer
    {
        using RpcServerPtr = std::shared_ptr<brpc::Server>;

    public:
        using ptr = std::shared_ptr<MessageServer>;
        MessageServer(const MQClient::ptr &mq_client,
                   const Discovery::ptr &dis_client,
                   const Registry::ptr &reg_client,
                   const RpcServerPtr &rpc_server,
                   const ElasticPtr &es_client,
                   const DBPtr &mysql_client)
            : _mq_client(mq_client),
              _dis_client(dis_client),
              _reg_client(reg_client),
              _rpc_server(rpc_server),
              _es_client(es_client),
              _mysql_client(mysql_client) {}
        ~MessageServer() {}

        // 启动RPC服务器
        void start()
        {
            _rpc_server->RunUntilAskedToQuit();
        }

    private:
        // 服务启动时对消息进行消费
        MQClient::ptr _mq_client; // 消息队列
        std::string _exchange_name;
        std::string _queue_name;

        // 项目运行期间都需要持续维护这些对象
        ElasticPtr _es_client;
        DBPtr _mysql_client;
        Discovery::ptr _dis_client;
        Registry::ptr _reg_client;
        RpcServerPtr _rpc_server;
    };

    /*建造者设计模式，封装对象的构造过程，防止用户在进行操作时，对未构造的对象进行操作*/

    class MessageServerBuilder
    {
        using RpcServerPtr = std::shared_ptr<brpc::Server>;

    public:
        void make_mq_object(const std::string &mq_user_name,
                            const std::string &mq_password,
                            const std::string &mq_host,
                            const std::string &exchange_name,
                            const std::string &queue_name,
                            const std::string &binding_key)
        {
            _mq_client = std::make_shared<MQClient>(mq_user_name, mq_password, mq_host);
            _mq_client->declareComponents(exchange_name, queue_name, binding_key);
            _exchange_name = exchange_name;
            _queue_name = queue_name;
            // 当前模块为消息管理模块，所以需要对聊天时产生的消息进行管理
            // 而这部分消息在消息转发模块中添加、发布到了消息队列中，所以我们需要消费这些消息
            // 又因为我们消费消息后需要在es、mysql进行操作
            // 所以我们在make_rpc_server确定了所需模块都已经创建再进行consume消费！！！
        }
        void make_mysql_object(const std::string &mysql_user_name,
                               const std::string &mysql_password,
                               const std::string &db_name,
                               const std::string &mysql_host,
                               int mysql_port,
                               const std::string &cset,
                               int conn_pool_num)
        {
            _mysql_client = ODBFactory::create(mysql_user_name, mysql_password, db_name,
                                               mysql_host, mysql_port, cset, conn_pool_num);
        }
        void make_es_object(const std::vector<std::string> &es_host_list)
        {
            _es_client = ElasticFactory::create(es_host_list);
        }
        // 构造服务发现和服务管理客户端
        void make_find_object(const std::string &reg_host,
                              const std::string &base_service_name,
                              const std::string &file_service_name,
                              const std::string &user_service_name)
        {
            _user_service_name = user_service_name;
            _file_service_name = file_service_name;
            _service_manager = std::make_shared<ServiceManager>();
            _service_manager->declareService(file_service_name);
            LOG_INFO("在用户服务中添加了文件管理子服务[{}]", file_service_name);
            _service_manager->declareService(user_service_name);
            LOG_INFO("在用户服务中添加了用户管理子服务[{}]", user_service_name);
            auto put_cb = std::bind(&bite_im::ServiceManager::onServiceOnline, _service_manager.get(),
                                    std::placeholders::_1, std::placeholders::_2);
            auto del_cb = std::bind(&bite_im::ServiceManager::onServiceOffLine, _service_manager.get(),
                                    std::placeholders::_1, std::placeholders::_2);
            _dis_client = std::make_shared<Discovery>(base_service_name, reg_host, put_cb, del_cb);
        }
        // 构造服务注册客户端对象
        void make_reg_object(const std::string &reg_host,
                             const std::string &service_name,
                             const std::string &access_host)
        {
            _reg_client = std::make_shared<Registry>(reg_host);
            _reg_client->registry(service_name, access_host);
        }
        // 搭建rpc服务器
        void make_rpc_server(uint16_t port, int32_t time_out, uint8_t num_threads)
        {
            // 防止需求对象未构造
            if (_mq_client.get() == nullptr)
            {
                LOG_ERROR("服务器建造失败，未初始化消息队列模块");
                abort();
            }
            if (_es_client.get() == nullptr)
            {
                LOG_ERROR("服务器建造失败，未初始化es客户端模块");
                abort();
            }
            if (_mysql_client.get() == nullptr)
            {
                LOG_ERROR("服务器建造失败，未初始化mysql客户端模块");
                abort();
            }
            if (_service_manager.get() == nullptr)
            {
                LOG_ERROR("服务器建造失败，未初始化服务管理模块");
                abort();
            }

            // 创建rpc服务器对象
            _rpc_server = std::make_shared<brpc::Server>();

            // 构建服务，并添加服务
            MessageServiceImpl *service = new MessageServiceImpl(_es_client, _mysql_client, _service_manager,
                                                                 _file_service_name, _user_service_name); // 注意这里需要在堆上开辟空间
            int ret = _rpc_server->AddService(service, brpc::ServiceOwnership::SERVER_OWNS_SERVICE);
            if (ret == -1)
            {
                LOG_ERROR("rpc服务器添加服务失败");
                abort();
            }

            auto options = std::make_shared<brpc::ServerOptions>();
            // 启动服务器
            options->idle_timeout_sec = time_out; // 连接空闲超时时间
            options->num_threads = num_threads;
            ret = _rpc_server->Start(port, options.get());
            if (ret == -1)
            {
                LOG_ERROR("服务器启动失败");
                abort();
            }

            // 构造完es、mysql和文件管理模块，开始进行消息队列的消费+
            auto callback = std::bind(&MessageServiceImpl::onMessage, service, std::placeholders::_1, std::placeholders::_2);
            _mq_client->consume(_queue_name, "consumer",callback);
        }

        MessageServer::ptr build()
        {
            // 检测维护对象是否建立
            if (_dis_client.get() == nullptr)
            {
                LOG_ERROR("服务器建造失败，未初始化服务发现模块");
                abort();
            }
            if (_reg_client.get() == nullptr)
            {
                LOG_ERROR("服务器建造失败，未初始化服务注册模块");
                abort();
            }
            if (_rpc_server.get() == nullptr)
            {
                LOG_ERROR("服务器建造失败，未搭建Rpc服务器");
                abort();
            }

            MessageServer::ptr server = std::make_shared<MessageServer>(_mq_client, _dis_client, _reg_client, _rpc_server,
                                                                        _es_client, _mysql_client);
            return server;
        }

    private:
        // 消息队列
        MQClient::ptr _mq_client;
        std::string _exchange_name;
        std::string _queue_name;

        // 数据管理模块
        ElasticPtr _es_client;
        DBPtr _mysql_client;

        // 服务注册和发现模块
        std::string _user_service_name;
        std::string _file_service_name;
        ServiceManager::ptr _service_manager;
        Discovery::ptr _dis_client;
        Registry::ptr _reg_client;

        // rpc服务器
        RpcServerPtr _rpc_server;
    };
}